package io.gitee.kenewstar.dbmq.handler;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import io.gitee.kenewstar.dbmq.ConsumerRegisterInfo;
import io.gitee.kenewstar.dbmq.ConsumerTask;
import io.gitee.kenewstar.dbmq.DbMqConfigProperties;
import io.gitee.kenewstar.dbmq.DbMqConsumer;
import io.gitee.kenewstar.dbmq.DbMqConsumerInf;
import io.gitee.kenewstar.dbmq.data.DbDataHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.stream.Collectors;

/**
 * @author kenewstar
 */
public abstract class DbMqCoreHandler
        implements InitializingBean, DbMqRegisterHandler, DbMqQueueHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(DbMqCoreHandler.class);

    protected DbDataHandler dbDataHandler;
    protected DbMqConfigProperties dbMqConfigProperties;


    protected void checkConsumerDuplicate(List<DbMqConsumerInf> infList, List<ConsumerTask> taskList) {
        Set<String> topics = new HashSet<>(infList.size());
        for (DbMqConsumerInf consumerInf : infList) {
            final DbMqConsumer consumer = consumerInf.getClass().getAnnotation(DbMqConsumer.class);
            if (Objects.isNull(consumer)) {
                continue;
            }
            final int size = topics.size();
            topics.add(consumer.topic());
            final int currentSize = topics.size();
            Assert.isTrue(size != currentSize, "topic[" + consumer.topic() + "] consumer duplicate");

            taskList.add(ConsumerTask.build(consumerInf, null, UUID.randomUUID().toString()));
        }
    }


    protected boolean checkRegisterExpire(ConsumerRegisterInfo info) {
        int hbInterval = dbMqConfigProperties.getHeartbeatInterval();
        if (hbInterval < dbMqConfigProperties.getPeriod()) {
            hbInterval = dbMqConfigProperties.getPeriod();
        }
        final DateTime heartbeatTime = DateUtil.offsetMillisecond(info.getHeartbeatTime(), hbInterval * 3);
        return DateUtil.compare(DateUtil.date(), heartbeatTime) > 0 || Objects.equals(info.getStatus(), "down");
    }


    @Override
    public void afterPropertiesSet() throws Exception {

    }


    @Override
    public void initRegisterInfo() {
        LOGGER.info("==========================初始化注册表信息=============================");
        this.dbDataHandler.initConsumerRegister();
    }


    @Override
    public void initQueueInfo(List<ConsumerTask> taskList) {
        LOGGER.info("==========================初始化队列信息===============================");
        this.dbDataHandler.initConsumerQueue(taskList);
    }


    @Override
    public void registerConsumer(List<ConsumerTask> taskList) {
        LOGGER.info("=================初始化注册消费者===================");

        final List<String> topics = taskList.stream().map(ConsumerTask::getTopic).collect(Collectors.toList());
        final List<ConsumerRegisterInfo> registerInfo = this.dbDataHandler.pullRegisterInfo(topics);

        final Map<String, List<ConsumerRegisterInfo>> topicGroup = registerInfo.stream().collect(Collectors.groupingBy(ConsumerRegisterInfo::getTopic));

        for (ConsumerTask task : taskList) {
            final List<ConsumerRegisterInfo> topicArrays = topicGroup.get(task.getTopic());
            if (Objects.isNull(topicArrays) || CollUtil.isEmpty(topicArrays)) {
                this.dbDataHandler.createConsumer(task);
                continue;
            }

            final List<ConsumerRegisterInfo> expireData = topicArrays.stream().filter(this::checkRegisterExpire).collect(Collectors.toList());
            if (CollUtil.isEmpty(expireData)) {
                this.dbDataHandler.createConsumer(task);
            } else {
                this.dbDataHandler.updateConsumer(task, expireData);
            }

            this.dbDataHandler.registerDown(expireData);
        }

    }


    @Override
    public List<ConsumerRegisterInfo> pullRegisterUpdate(List<ConsumerTask> taskList) {
        LOGGER.debug("=================拉取注册信息===================");

        final List<String> topics = taskList.stream().map(ConsumerTask::getTopic).collect(Collectors.toList());
        final List<ConsumerRegisterInfo> registerInfo = this.dbDataHandler.pullRegisterInfo(topics);
        final Map<String, List<ConsumerRegisterInfo>> topicGroup = registerInfo.stream().collect(Collectors.groupingBy(ConsumerRegisterInfo::getTopic));

        for (ConsumerTask task : taskList) {
            final List<ConsumerRegisterInfo> topicArrays = topicGroup.get(task.getTopic());
            if (Objects.isNull(topicArrays) || CollUtil.isEmpty(topicArrays)) {
                registerInfo.add(this.dbDataHandler.createConsumer(task));
                continue;
            }

            final long count = topicArrays.stream().filter(x -> Objects.equals(task.getId(), x.getId())).count();
            if (count == 0) {
                registerInfo.add(this.dbDataHandler.createConsumer(task));
                continue;
            }

            final List<ConsumerRegisterInfo> expireData = topicArrays.stream().filter(this::checkRegisterExpire).collect(Collectors.toList());
            if (CollUtil.isNotEmpty(expireData) && Objects.equals(topicArrays.get(topicArrays.size() - 1).getId(), task.getId())) {
                this.dbDataHandler.updateConsumer(task, expireData);
                this.dbDataHandler.registerDown(task.getOldId());
            }

            for (int i = 0; i < topicArrays.size(); i++) {
                if (Objects.equals(topicArrays.get(i).getId(), task.getId())) {
                    task.setTag(i + 1);
                    break;
                }
            }
        }

        return registerInfo;

    }
}
